// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: discovery support client and server
// ------------------------------------------------------------------

#include "FlowControllerFactory.hpp"
#include "FlowControllerImpl.hpp"

#include <elog/Log.hpp>

namespace evbs {
namespace edds {
namespace rtps {

void FlowControllerFactory::init(ertps::rtps::RTPSParticipantImpl* participant) {
    participant_ = participant;
    // Create default flow controllers.

    // PureSyncFlowController -> used by volatile besteffort writers.
    (void)flow_controllers_.insert(std::make_pair<std::string, std::unique_ptr<FlowController>>(
        pure_sync_flow_controller_name,
        std::unique_ptr<FlowController>(
            new FlowControllerImpl<FlowControllerPureSyncPublishMode, FlowControllerFifoSchedule>(participant_,
                                                                                                  nullptr))));
    // SyncFlowController -> used by rest of besteffort writers.
    (void)flow_controllers_.insert(std::make_pair<std::string, std::unique_ptr<FlowController>>(
        sync_flow_controller_name,
        std::unique_ptr<FlowController>(
            new FlowControllerImpl<FlowControllerSyncPublishMode, FlowControllerFifoSchedule>(participant_, nullptr))));
    // AsyncFlowController
    (void)flow_controllers_.insert(std::make_pair<std::string, std::unique_ptr<FlowController>>(
        async_flow_controller_name,
        std::unique_ptr<FlowController>(
            new FlowControllerImpl<FlowControllerAsyncPublishMode, FlowControllerFifoSchedule>(participant_,
                                                                                               nullptr))));
}

void FlowControllerFactory::register_flow_controller(const FlowControllerDescriptor& flow_controller_descr) {
    if (flow_controllers_.end() != flow_controllers_.find(flow_controller_descr.name)) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ILLEGAL_OPERATION,
                  "Error registering FlowController " << flow_controller_descr.name << ". Already registered");
        return;
    }

    if (0 < flow_controller_descr.max_bytes_per_period) {
        switch (flow_controller_descr.scheduler) {
            case FlowControllerSchedulerPolicy::FIFO:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(
                         new FlowControllerImpl<FlowControllerLimitedAsyncPublishMode, FlowControllerFifoSchedule>(
                             participant_, &flow_controller_descr))});
                break;
            case FlowControllerSchedulerPolicy::ROUND_ROBIN:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(new FlowControllerImpl<FlowControllerLimitedAsyncPublishMode,
                                                                            FlowControllerRoundRobinSchedule>(
                         participant_, &flow_controller_descr))});
                break;
            case FlowControllerSchedulerPolicy::HIGH_PRIORITY:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(new FlowControllerImpl<FlowControllerLimitedAsyncPublishMode,
                                                                            FlowControllerHighPrioritySchedule>(
                         participant_, &flow_controller_descr))});
                break;
            case FlowControllerSchedulerPolicy::PRIORITY_WITH_RESERVATION:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(
                         new FlowControllerImpl<FlowControllerLimitedAsyncPublishMode,
                                                FlowControllerPriorityWithReservationSchedule>(
                             participant_, &flow_controller_descr))});
                break;
            default:
                assert(false);
        }
    } else {
        switch (flow_controller_descr.scheduler) {
            case FlowControllerSchedulerPolicy::FIFO:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(
                         new FlowControllerImpl<FlowControllerAsyncPublishMode, FlowControllerFifoSchedule>(
                             participant_, &flow_controller_descr))});
                break;
            case FlowControllerSchedulerPolicy::ROUND_ROBIN:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(
                         new FlowControllerImpl<FlowControllerAsyncPublishMode, FlowControllerRoundRobinSchedule>(
                             participant_, &flow_controller_descr))});
                break;
            case FlowControllerSchedulerPolicy::HIGH_PRIORITY:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(
                         new FlowControllerImpl<FlowControllerAsyncPublishMode, FlowControllerHighPrioritySchedule>(
                             participant_, &flow_controller_descr))});
                break;
            case FlowControllerSchedulerPolicy::PRIORITY_WITH_RESERVATION:
                (void)flow_controllers_.insert(
                    {flow_controller_descr.name,
                     std::unique_ptr<FlowController>(
                         new FlowControllerImpl<FlowControllerAsyncPublishMode,
                                                FlowControllerPriorityWithReservationSchedule>(
                             participant_, &flow_controller_descr))});
                break;
            default:
                assert(false);
        }
    }
}

/*!
 * Get a FlowController given its name.
 *
 * @param flow_controller_name Name of the interested FlowController.
 * @return Pointer to the FlowController. nullptr if no registered FlowController with that name.
 */
FlowController* FlowControllerFactory::retrieve_flow_controller(
    const std::string& flow_controller_name, const ertps::rtps::WriterAttributes& writer_attributes) {
    FlowController* returned_flow = nullptr;

    // Detect it has to be returned a default flow_controller.
    if (0 == flow_controller_name.compare(vbsutil::flowctlString::EDDS_FLOW_CONTROLLER_DEFAULT)) {
        if (ertps::rtps::SYNCHRONOUS_WRITER == writer_attributes.mode) {
            if (ertps::rtps::BEST_EFFORT == writer_attributes.endpoint.reliabilityKind) {
                returned_flow = flow_controllers_[pure_sync_flow_controller_name].get();
            } else {
                returned_flow = flow_controllers_[sync_flow_controller_name].get();
            }
        } else {
            returned_flow = flow_controllers_[async_flow_controller_name].get();
        }
    } else {
        auto it = flow_controllers_.find(flow_controller_name);

        if (flow_controllers_.end() != it) {
            returned_flow = it->second.get();
        }
    }

    if (nullptr != returned_flow) {
        returned_flow->init();
    } else {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                  "Cannot find FlowController " << flow_controller_name << ".");
    }

    return returned_flow;
}

}  // namespace rtps
}  // namespace edds
}  // namespace evbs
